package com.lxb.demo02

import org.apache.flink.streaming.api.scala._

/**
 * @author: albert
 * @date: 2021/5/20 15:21
 * @description:
 */
object ConnectDemo {

//  def main(args:Array[String]):Unit = {
//     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//
//    // 文件地址
//    val input = "D:\\project\\bigdata\\demoall\\flinkDemo\\src\\main\\resources\\sensor.txt"
//
//     val dataStream: DataStream[Sensor] = env.readTextFile(input)
//      .map(data => {
//        val strings: Array[String] = data.split(",")
//        Sensor(strings(0).trim(), strings(1).trim.toLong, strings(2).trim.toDouble)
//      })
//
//     val splitData: SplitStream[Sensor] = dataStream.split(data => {
//      if (data.temporary > 30) Seq("high")
//      else
//        Seq("low")
//    })
//
//     val waringData: DataStream[(String, Long)] = splitData.select("high").map(data => (data.name, data.timestamp))
//
//     val lowData: DataStream[Sensor] = splitData.select("low")
//
//     val connectData: DataStream[(String, Double, String)] = lowData.connect(waringData).map(
//      lowerData => (lowerData.name, lowerData.temporary, "low"),
//      warnData => (warnData._1, warnData._2, "warn")
//    )
//
//    connectData.print("connectData")
//
//    env.execute()
//  }
}
case class Sensor(name: String,timestamp: Long,temporary:Double)